package com.atguigu.flink.exec1;

import com.atguigu.flink.pojo.UserBehavior;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashSet;
import java.util.Set;

/**
 * Created by Smexy on 2022/12/14
 *
 *      1.读数据，获取Source，把原始数据封装为POJO
 *      2.计算
 *      3.输出
 *
 *      (pv,434349)
 */
public class Demo2_UV
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setRuntimeMode(RuntimeExecutionMode.BATCH);
        
                env
                   .readTextFile("data/UserBehavior.csv")
                   .map(new MapFunction<String, UserBehavior>()
                   {
                       @Override
                       public UserBehavior map(String value) throws Exception {
                           String[] words = value.split(",");
                           UserBehavior userBehavior = new UserBehavior(
                               Long.valueOf(words[0]),
                               Long.valueOf(words[1]),
                               Integer.valueOf(words[2]),
                               words[3],
                               Long.valueOf(words[4])
                           );
                           return userBehavior;
                       }
                   })
                   .filter(u -> "pv".equals(u.getBehavior()))
                   .map(u -> u.getUserId())
                   .process(new ProcessFunction<Long, Integer>()
                   {
                       Set<Long> uids = new HashSet<>();
                       /*
                            1
                            1
                            1
                        */
                       @Override
                       public void processElement(Long id, Context ctx, Collector<Integer> out) throws Exception {

                           //返回true，说明添加成功，返回false，当前用户已经重复
                           if (uids.add(id)) out.collect(uids.size());
                       }
                   }).setParallelism(1)
                   .print().setParallelism(1);

        
                try {
                            env.execute();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }

    }
}
